//
// Created by 33145 on 2023/6/9.
//

#include <google/protobuf/descriptor.h>
#include <google/protobuf/message.h>

#include "rpc_channel.h"
#include "../coder/tinypb_protocol.h"
#include "../common/rocket.h"
#include "../rpc/rpc_controller.h"
#include "../tcp/tcp_client.h"
#include "../common/err_code.h"
#include "../net/timer_event.h"
#include "../net/timer.h"

namespace rocket {

  void
  RPCChannel::CallMethod(const google::protobuf::MethodDescriptor *method, google::protobuf::RpcController *controller,
                         const google::protobuf::Message *request, google::protobuf::Message *response,
                         google::protobuf::Closure *done) {

    TinyPBProtocol::ptr req_protocol = std::make_shared<TinyPBProtocol>();  // 要发送的protocol
    auto* my_controller = dynamic_cast<RPCController*>(controller);
    if (my_controller == nullptr) {
      LOG_ERROR("faild callmethod, RPCController convert error");
      return;
    }

    if (my_controller->GetMsgId().empty()) {
      req_protocol->_msg_id = MsgIDUtil::GenMsgID();
      my_controller->SetMsgId(req_protocol->_msg_id);
    } else {
      req_protocol->_msg_id = my_controller->GetMsgId();
    }

    req_protocol->_method_name = method->full_name();
    LOG_INFO("%s | call method name [%s]", req_protocol->_msg_id.c_str(), req_protocol->_method_name.c_str());

    if (!_is_init) {
      std::string err = "RPCChannel not init";
      my_controller->SetError(ERROR_RPC_CHANNEL_INIT, err);
      LOG_ERROR("%s | RPCChannel not init [%s] ", req_protocol->_msg_id.c_str(), request->ShortDebugString().c_str());
      return;
    }

    // request 序列化请求
    if (!request->SerializeToString(&(req_protocol->_pb_data))) {
      std::string err = "failed SerializeToString request";
      my_controller->SetError(ERROR_FAILED_SERIALIZE, err);
      LOG_ERROR("%s | %s, origin request [%s] ", req_protocol->_msg_id.c_str(), err.c_str(), request->ShortDebugString().c_str());
    }

    ptr channel = shared_from_this();

    _timer_event = std::make_shared<TimerEvent>(my_controller->GetTimeout(), [my_controller, channel]() mutable {
      // 超时时，取消RPC
      if (my_controller->Finished()) {
        channel.reset();
        return;
      }

      my_controller->StartCancel();
      my_controller->SetError(ERROR_RPC_CALL_TIMEOUT, "rpc call timeout " + std::to_string(my_controller->GetTimeout()));

      if (channel->getClosure()) {
        channel->getClosure()->Run();
      }
      channel.reset();
    });


    _client->addTimerEvent(_timer_event);

    // 丑陋啊。。。
    _client->connect([req_protocol, channel]() {

      if (channel->getTCPClient()->getConnectErrorCode() != 0) {
        RPCController::ptr my_controller = std::dynamic_pointer_cast<RPCController>(channel->getController());
        my_controller->SetError(ERROR_FAILED_CONNECT, "connect error");
        LOG_ERROR("connection error msg_id [%S]  [%s] peer_addr [%s]", req_protocol->_msg_id.c_str(),
                  my_controller->GetErrorInfo().c_str(), channel->getTCPClient()->getPeerAddr()->toString().c_str());
        return;
      }

      channel->getTCPClient()->writeMessage(req_protocol, [channel](const AbstractProtocol::ptr& message) {
        auto req_protocol = std::dynamic_pointer_cast<TinyPBProtocol>(message);
        LOG_INFO("%s | send request success method name[%s]", req_protocol->_msg_id.c_str(),
                 req_protocol->_method_name.c_str());
        // 为什么要加了mutable才能修改channel
        channel->getTCPClient()->readMessage(req_protocol->_msg_id, [req_protocol, channel](const AbstractProtocol::ptr& message) mutable {

          // rpc成功，取消定时器
          channel->getTimerEvent()->setCancled(true);

          auto rsp_protocol = std::dynamic_pointer_cast<TinyPBProtocol>(message);
          LOG_INFO("%s | send response success method name[%s]", rsp_protocol->_msg_id.c_str(),
                   req_protocol->_method_name.c_str());

          if (!channel->getResponse()->ParseFromString(rsp_protocol->_pb_data)) {
            LOG_ERROR("%s | SerializeToString error [%s]", rsp_protocol->_msg_id.c_str(),
                      req_protocol->_method_name.c_str());
            RPCController::ptr my_controller = std::dynamic_pointer_cast<RPCController>(channel->getController());
            my_controller->SetError(ERROR_FAILED_DESERIALIZE, "SerializeToString error");
            return;
          }
          if (rsp_protocol->_err_code != 0) {
            LOG_ERROR("%s | call rpc failed error code [%d], error info [%s]", rsp_protocol->_msg_id.c_str() ,
                      rsp_protocol->_err_code, rsp_protocol->_err_info.c_str());
            RPCController::ptr my_controller = std::dynamic_pointer_cast<RPCController>(channel->getController());
            my_controller->SetError(rsp_protocol->_err_code, "call rpc failed");
          }

          LOG_INFO("[%s] rpc call success method_name [%s] peer_addr [%s]", rsp_protocol->_msg_id.c_str(),req_protocol->_method_name.c_str(), channel->getTCPClient()->getPeerAddr()->toString().c_str());


          if (channel->getClosure()) {
            channel->getClosure()->Run();
          }
          channel.reset();
        });
      });
    });
  }

  RPCChannel::RPCChannel(NetAddr::ptr peer_addr) : _peer_addr(peer_addr) {
    _client = std::make_shared<TCPClient>(_peer_addr);
  }

  void RPCChannel::Init(const RPCChannel::controller_ptr& controller, const RPCChannel::message_ptr& req,
                        const RPCChannel::message_ptr& res,
                        const RPCChannel::closure_ptr& done) {
    if (_is_init) {
      return;
    }
    _controller = controller;
    _request = req;
    _response = res;
    _closure = done;
    _is_init = true;
  }

  RPCChannel::~RPCChannel() {
    LOG_DEBUG("~RPCChannel ");
  }

  TCPClient::ptr RPCChannel::getTCPClient() {
    return _client;
  }

  TimerEvent::ptr RPCChannel::getTimerEvent() {
    return _timer_event;
  }
}